// Copyright 2016 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------
// Modification history:
// feature: change E2E packet format
// feature: Support the reader to filter again through the writer's filtering results
// feature: develop dispatcher
// feature: content filter topic
// feature: Split history as an independent common module
// feature: log event record topic name
// feature: add statictic lookup
// feature: E2E Profile04 and Configuration Function Development
// ------------------------------------------------------------------

#ifndef INCLUDE_EDDS_RTPS_WRITER_RTPSWRITER_H_
#define INCLUDE_EDDS_RTPS_WRITER_RTPSWRITER_H_

#include <history/SubstrateWriter.h>
#include <edds/rtps/Endpoint.h>
#include <edds/rtps/attributes/HistoryAttributes.h>
#include <edds/rtps/attributes/WriterAttributes.h>
#include <edds/rtps/builtin/data/ReaderProxyData.h>
#include <edds/rtps/messages/RTPSMessageGroup.h>
#include <ertps/qos/LivelinessLostStatus.h>

#include <chrono>
#include <functional>
#include <memory>
#include <mutex>
#include <vector>
#include <edds/dds/builtin/lookup/common/LookupTypes.hpp>
#include <elog/statistics/Statistics.hpp>
#include <edds/rtps/interfaces/IReaderDataFilter.hpp>

#include "DeliveryRetCode.hpp"
#include "LocatorSelectorSender.hpp"
#include <history/ReaderHistory.h>
#include <history/WriterHistory.h>
#include <history/WriterListener.h>

namespace evbs {

namespace edds {
namespace rtps {

class FlowController;

}  // namespace rtps

namespace dds {

class DataWriter;

}  // namespace dds
}  // namespace edds

namespace ertps {
namespace rtps {
using vbs::common::CacheChange_t;
using vbs::common::WriterHistory;
using vbs::common::WriterListener;

/**
 * Class RTPSWriter, manages the sending of data to the readers. Is always associated with a
 * HistoryCache.
 * @ingroup WRITER_MODULE
 */
class RTPSWriter : public Endpoint,
                   public vbsutil::statistics::StatisticsWriter,
                   public vbs::dispatcher::SubstrateWriter {
    //friend class WriterHistory;
    friend class RTPSParticipantImpl;
    friend class RTPSMessageGroup;
    friend class edds::dds::DataWriter;

 protected:
    RTPSWriter(RTPSParticipantImpl* impl, const GUID_t& guid, const WriterAttributes& att,
               edds::rtps::FlowController* flow_controller, WriterHistory* hist, WriterListener* listen = nullptr);

    RTPSWriter(RTPSParticipantImpl* impl, const GUID_t& guid, const WriterAttributes& att,
               const std::shared_ptr<vbs::common::IPayloadPool>& payload_pool,
               edds::rtps::FlowController* flow_controller, WriterHistory* hist, WriterListener* listen = nullptr);

    RTPSWriter(RTPSParticipantImpl* impl, const GUID_t& guid, const WriterAttributes& att,
               const std::shared_ptr<vbs::common::IPayloadPool>& payload_pool,
               const std::shared_ptr<vbs::common::IChangePool>& change_pool,
               edds::rtps::FlowController* flow_controller, WriterHistory* hist, WriterListener* listen = nullptr);

    virtual ~RTPSWriter();

 public:
    /**
     * Create a new change based with the provided changeKind.
     * @param data Data of the change.
     * @param changeKind The type of change.
     * @param handle InstanceHandle to assign.
     * @return Pointer to the CacheChange or nullptr if incorrect.
     */

    template <typename T>
    CacheChange_t* new_change(T& data, ChangeKind_t changeKind, InstanceHandle_t handle = c_InstanceHandle_Unknown) {
        // 将捕获的引用改为值传递
        return new_change([data]() -> uint32_t { return (uint32_t)T::getCdrSerializedSize(data); }, changeKind, handle);
    }

    RTPS_DllAPI CacheChange_t* new_change(const std::function<uint32_t()>& dataCdrSerializedSize,
                                          ChangeKind_t changeKind, InstanceHandle_t handle = c_InstanceHandle_Unknown);

    RTPS_DllAPI CacheChange_t* new_change(ChangeKind_t changeKind, InstanceHandle_t handle = c_InstanceHandle_Unknown);

    /**
     * Release a change when it is not being used anymore.
     *
     * @param change Pointer to the cache change to be released.
     *
     * @returns whether the operation succeeded or not
     *
     * @pre
     *     @li @c change is not @c nullptr
     *     @li @c change points to a cache change obtained from a call to @c this->new_change
     *
     * @post memory pointed to by @c change is not accessed
     */
    RTPS_DllAPI bool release_change(CacheChange_t* change);

    /**
     * Add a matched reader.
     * @param data Pointer to the ReaderProxyData object added.
     * @return True if added.
     */
    RTPS_DllAPI virtual bool matched_reader_add(ReaderProxyData& data) = 0;

    /**
     * Remove a matched reader.
     * @param reader_guid GUID of the reader to remove.
     * @return True if removed.
     */
    RTPS_DllAPI virtual bool matched_reader_remove(const GUID_t& reader_guid) = 0;

    void add_announce_locator(Locator_t loc) override;

    /**
     * Tells us if a specific Reader is matched against this writer.
     * @param reader_guid GUID of the reader to check.
     * @return True if it was matched.
     */
    RTPS_DllAPI virtual bool matched_reader_is_matched(const GUID_t& reader_guid) = 0;

    RTPS_DllAPI virtual size_t getMatchedReadersSize() const = 0;

    /**
     * @return The content filter associated to this reader.
     */
    RTPS_DllAPI IReaderDataFilter* reader_data_filter() {
        std::lock_guard<vbs::common::RecursiveTimedMutex> lock(mp_mutex);
        return data_filter_;
    }

    /**
     * Set the content filter associated to this reader.
     *
     * @param filter  Pointer to the content filter to associate to this reader.
     */
    RTPS_DllAPI void reader_data_filter(IReaderDataFilter* filter) {
        std::lock_guard<vbs::common::RecursiveTimedMutex> lock(mp_mutex);
        data_filter_ = filter;
    }

    /**
     * @brief Check if a specific change has been delivered to the transport layer of every matched remote RTPSReader
     * at least once.
     *
     * @param seq_num Sequence number of the change to check.
     * @return true if delivered. False otherwise.
     */
    RTPS_DllAPI virtual bool has_been_fully_delivered(const SequenceNumber_t& seq_num) const {  // LCOV_EXCL_START
        static_cast<void>(seq_num);
        return false;
    }

    /**
     * Check if a specific change has been acknowledged by all Readers.
     * Is only useful in reliable Writer. In BE Writers returns false when pending to be sent.
     * @return True if acknowledged by all.
     */
    RTPS_DllAPI virtual bool is_acked_by_all(const CacheChange_t* /*a_change*/) const {
        return false;
    }  // LCOV_EXCL_STOP

    /**
     * Waits until all changes were acknowledged or max_wait.
     * @return True if all were acknowledged.
     */
    RTPS_DllAPI virtual bool wait_for_all_acked(const Duration_t& /*max_wait*/) { return true; }

    /**
     * Update the Attributes of the Writer.
     * @param att New attributes
     */
    RTPS_DllAPI virtual void updateAttributes(const WriterAttributes& att) = 0;

    RTPS_DllAPI virtual edds::dds::builtin::StatisticMatchGuids get_remote_guids() {  // LCOV_EXCL_START
        return edds::dds::builtin::StatisticMatchGuids();
    };

    RTPS_DllAPI virtual edds::dds::builtin::StatisticProxyInfos get_proxy_infos() {
        return edds::dds::builtin::StatisticProxyInfos();
    };  // LCOV_EXCL_STOP

    /**
     * Get Min Seq Num in History.
     * @return Minimum sequence number in history
     */
    RTPS_DllAPI SequenceNumber_t get_seq_num_min();

    /**
     * Get Max Seq Num in History.
     * @return Maximum sequence number in history
     */
    RTPS_DllAPI SequenceNumber_t get_seq_num_max();

    /**
     * Get Minimum sequence number from minimum batch history.
     * @return Minimum sequence number from minimum batch history
     */
    RTPS_DllAPI SequenceNumber_t get_last_seq_from_min_batch();

    /**
     * Get maximum size of the serialized type
     * @return Maximum size of the serialized type
     */
    RTPS_DllAPI uint32_t getTypeMaxSerialized();

    uint32_t getInlineQosSize(const CacheChange_t* change);

    //! Get maximum size of the data
    uint32_t getMaxDataSize(const CacheChange_t* change);

    //! Calculates the maximum size of the data
    uint32_t calculateMaxDataSize(uint32_t length);

    /**
     * Get listener
     * @return Listener
     */
    RTPS_DllAPI inline WriterListener* getListener() { return mp_listener; }
    //user writer is't able to call this func
    RTPS_DllAPI inline bool set_listener(WriterListener* const listener);

    /**
     * Get the publication mode
     * @return publication mode
     */
    RTPS_DllAPI inline bool isAsync() const { return is_async_; }

    /**
     * Remove an specified max number of changes
     * @param max Maximum number of changes to remove.
     * @return at least one change has been removed
     */
    RTPS_DllAPI bool remove_older_changes(uint32_t max = 0U);

    /**
     * @brief Returns if disable positive ACKs QoS is enabled.
     *
     * @return Best effort writers always return false.
     *         Reliable writers override this method.
     */
    RTPS_DllAPI virtual bool get_disable_positive_acks() const { return false; }

    /**
     * Tries to remove a change waiting a maximum of the provided microseconds.
     * @param max_blocking_time_point Maximum time to wait for.
     * @param lock Lock of the Change list.
     * @return at least one change has been removed
     */
    virtual bool try_remove_change(
        const std::chrono::steady_clock::time_point& max_blocking_time_point,  // LCOV_EXCL_START
        std::unique_lock<vbs::common::RecursiveTimedMutex>& lock) {
        UNUSED_PARAMETER(max_blocking_time_point);
        UNUSED_PARAMETER(lock);
        return true;
    }

    /**
     * Waits till a change has been acknowledged.
     * @param seq Sequence number to wait for acknowledgement.
     * @param max_blocking_time_point Maximum time to wait for.
     * @param lock Lock of the Change list.
     * @return true when change was acknowledged, false when timeout is reached.
     */
    virtual bool wait_for_acknowledgement(const SequenceNumber_t& seq,
                                          const std::chrono::steady_clock::time_point& max_blocking_time_point,
                                          std::unique_lock<vbs::common::RecursiveTimedMutex>& lock) {
        UNUSED_PARAMETER(seq);
        UNUSED_PARAMETER(max_blocking_time_point);
        UNUSED_PARAMETER(lock);
        return true;
    }  // LCOV_EXCL_STOP

    bool change_is_acked_or_fully_delivered(const CacheChange_t* change) override;
    /**
     * Get RTPS participant
     * @return RTPS participant
     */
    inline RTPSParticipantImpl* getRTPSParticipant() const { return mp_RTPSParticipant; }

    /**
     * Enable or disable sending data to readers separately
     * NOTE: This will only work for synchronous writers
     * @param enable If separate sending should be enabled
     */
    void set_separate_sending(const bool enable) { m_separateSendingEnabled = enable; }

    /**
     * Inform if data is sent to readers separately
     * @return true if separate sending is enabled
     */
    bool get_separate_sending() const { return m_separateSendingEnabled; }

    /**
     * Process an incoming ACKNACK submessage.
     * @param[in] writer_guid      GUID of the writer the submessage is directed to.
     * @param[in] reader_guid      GUID of the reader originating the submessage.
     * @param[in] ack_count        Count field of the submessage.
     * @param[in] sn_set           Sequence number bitmap field of the submessage.
     * @param[in] final_flag       Final flag field of the submessage.
     * @param[out] result          true if the writer could process the submessage.
     *                             Only valid when returned value is true.
     * @return true when the submessage was destinated to this writer, false otherwise.
     */
    virtual bool process_acknack(const GUID_t& writer_guid, const GUID_t& reader_guid,
                                 uint32_t ack_count,  // LCOV_EXCL_START
                                 const SequenceNumberSet_t& sn_set, const bool final_flag, bool& result) {
        (void)reader_guid;
        (void)ack_count;
        (void)sn_set;
        (void)final_flag;

        result = false;
        return writer_guid == m_guid;
    }

    /**
     * Process an incoming NACKFRAG submessage.
     * @param[in] writer_guid      GUID of the writer the submessage is directed to.
     * @param[in] reader_guid      GUID of the reader originating the submessage.
     * @param[in] ack_count        Count field of the submessage.
     * @param[in] seq_num          Sequence number field of the submessage.
     * @param[in] fragments_state  Fragment number bitmap field of the submessage.
     * @param[out] result          true if the writer could process the submessage.
     *                             Only valid when returned value is true.
     * @return true when the submessage was destinated to this writer, false otherwise.
     */
    virtual bool process_nack_frag(const GUID_t& writer_guid, const GUID_t& reader_guid, uint32_t ack_count,
                                   const SequenceNumber_t& seq_num, const FragmentNumberSet_t fragments_state,
                                   bool& result) {
        (void)reader_guid;
        (void)ack_count;
        (void)seq_num;
        (void)fragments_state;

        result = false;
        return writer_guid == m_guid;
    }  // LCOV_EXCL_STOP

    /**
     * @brief A method to retrieve the liveliness kind
     *
     * @return Liveliness kind
     */
    const LivelinessQosPolicyKind& get_liveliness_kind() const;

    /**
     * @brief A method to retrieve the liveliness lease duration
     *
     * @return Lease duration
     */
    const Duration_t& get_liveliness_lease_duration() const;

    /**
     * @brief A method to return the liveliness announcement period
     *
     * @return The announcement period
     */
    const Duration_t& get_liveliness_announcement_period() const;

    //! Liveliness lost status of this writer
    LivelinessLostStatus liveliness_lost_status_;

    /*!
     * Tells writer the sample can be sent to the network.
     * This function should be used by a edds::rtps::FlowController.
     *
     * @param cache_change Pointer to the CacheChange_t that represents the sample which can be
     * sent.
     * @param group RTPSMessageGroup reference uses for generating the RTPS message.
     * @param locator_selector RTPSMessageSenderInterface reference uses for selecting locators. The
     * reference has to be a member of this RTPSWriter object.
     * @param max_blocking_time Future timepoint where blocking send should end.
     * @return Return code.
     * @note Must be non-thread safe.
     */
    virtual DeliveryRetCode deliver_sample_nts(
        CacheChange_t* cache_change, RTPSMessageGroup& group, LocatorSelectorSender& locator_selector,
        const std::chrono::time_point<std::chrono::steady_clock>& max_blocking_time, bool in_history = true) = 0;

    virtual LocatorSelectorSender& get_general_locator_selector() = 0;

    virtual LocatorSelectorSender& get_async_locator_selector() = 0;

    /**
     * Send a message through this interface.
     *
     * @param message Pointer to the buffer with the message already serialized.
     * @param locator_selector RTPSMessageSenderInterface reference uses for selecting locators. The
     * reference has to be a member of this RTPSWriter object.
     * @param max_blocking_time_point Future timepoint where blocking send should end.
     */
    virtual bool send_nts(CDRMessage_t* message, const LocatorSelectorSender& locator_selector) const;

    const std::string& getTopicName() const { /* override */
        return m_topicName;
    }

    void setTopicName(const std::string topicName) { m_topicName = topicName; }

    std::string getIdlCrc16() { return m_crc16Idl; }

    void setIdlCrc16(std::string crc16_input) { m_crc16Idl = std::move(crc16_input); }

    //! get is_send_multi
    bool getSendMulti() const { return is_send_multi; }

    //! set is_send_multi
    void setSendMulti(bool send_multi) { is_send_multi = send_multi; }

    bool write(vbs::common::CacheChange_t* data,
               const std::chrono::time_point<std::chrono::steady_clock>& max_blocking_time, bool in_history) override {
        unsent_change_added_to_history(data, max_blocking_time, in_history);
        return true;
    }

    uint8_t getTracePointStatus() const { return trace_point_status_; }

 protected:
    //! Flow controller.
    edds::rtps::FlowController* flow_controller_;

    //! WriterHistory
    WriterHistory* mp_history = nullptr;
    //! Listener
    WriterListener* mp_listener = nullptr;
    //! Asynchronous publication activated
    bool is_async_ = false;
    //! Separate sending activated, default false, but true with security
    bool m_separateSendingEnabled = false;

    //! The liveliness kind of this writer
    LivelinessQosPolicyKind liveliness_kind_;
    //! The liveliness lease duration of this writer
    Duration_t liveliness_lease_duration_;
    //! The liveliness announcement period
    Duration_t liveliness_announcement_period_;

    //! E2E Profile 04 information
    bool enable_e2e_protection_;
    bool topic_with_key_ = false;

    //! E2E Profile04
    uint16_t e2e_p04_min_data_length_;
    uint16_t e2e_p04_max_data_length_;
    uint16_t e2e_p04_max_delta_counter_;

    //! Whether sending multiple copies is supported
    bool is_send_multi = false;
    //! Whether builtin writer.
    bool builtin = false;
    E2EProfile04_t e2e_profile04_;

    uint8_t trace_point_status_;

    void add_guid(LocatorSelectorSender& locator_selector, const GUID_t& remote_guid);

    void compute_selected_guids(LocatorSelectorSender& locator_selector);

    void update_cached_info_nts(LocatorSelectorSender& locator_selector);

 public:
    /**
     * Add a change to the unsent list.
     * @param change Pointer to the change to add.
     * @param max_blocking_time
     */
    virtual void unsent_change_added_to_history(
        CacheChange_t* change, const std::chrono::time_point<std::chrono::steady_clock>& max_blocking_time,
        bool in_history) = 0;

 protected:
    /**
     * Indicate the writer that a change has been removed by the history due to some HistoryQos
     * requirement.
     * @param a_change Pointer to the change that is going to be removed.
     * @return True if removed correctly.
     */
    virtual bool change_removed_by_history(CacheChange_t* change) {  // LCOV_EXCL_START
        UNUSED_PARAMETER(change);
        return true;
    }  // LCOV_EXCL_STOP

    bool is_pool_initialized() const;

    void deinit();

 private:
    vbs::common::RecursiveTimedMutex& get_mutex() { return mp_mutex; }

    RTPSWriter& operator=(const RTPSWriter&) = delete;

    void init(const std::shared_ptr<vbs::common::IPayloadPool>& payload_pool,
              const std::shared_ptr<vbs::common::IChangePool>& change_pool, const WriterAttributes& att);

    // RTPSWriter* next_[2] = {nullptr, nullptr};

    std::string m_crc16Idl;

    std::string m_topicName;

    IReaderDataFilter* data_filter_ {};
};

} /* namespace rtps */
} /* namespace ertps */
} /* namespace evbs */

#endif  // INCLUDE_EDDS_RTPS_WRITER_RTPSWRITER_H_
